package com.fdb.efp.basic.framework.mq.client.producer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @description:消息发送客户端处理类
 * @date:2019年2月12日 上午9:19:03
 * @author:wangyk3
 */
public class MQProducerMessageClient {
	private static Logger logger = LoggerFactory.getLogger(MQProducerMessageClient.class);
	private DefaultMQProducer producer;
	private String producerGroup;
	private String namesrvAddr;
	private String retryAnotherBrokerWhenNotStoreOK;

	/**
	 * @description:初始化消息发送客户端
	 * @throws MQClientException
	 * @date:2019年2月12日 上午9:19:28
	 * @author:wangyk3
	 */
	public void init() throws MQClientException {
		logger.info("Producer initialize!!" + producerGroup + namesrvAddr);
		producer = new DefaultMQProducer(producerGroup);
		producer.setNamesrvAddr(namesrvAddr);
		producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
		producer.start();
		logger.info("Producer start success!");
	}

	/**
	 * @description:关闭消息发送客户端
	 * @date:2019年2月12日 上午9:20:22
	 * @author:wangyk3
	 */
	public void desdory() {
		producer.shutdown();
	}

	/**
	 * @description:发送消息
	 * @param msg
	 * @return
	 * @date:2019年2月12日 上午9:20:57
	 * @author:wangyk3
	 */
	public void sendMsg(Message msg) {
		SendResult sendResult = null;
		try {
			sendResult = producer.send(msg);
			logger.info(sendResult.toString());
		} catch (InterruptedException e) {
			logger.error("InterruptedException" + e.getMessage() + String.valueOf(sendResult));
		} catch (RemotingException e) {
			logger.error("RemotingException" + e.getMessage() + String.valueOf(sendResult));
		} catch (MQClientException e) {
			logger.error("MQClientException" + e.getMessage() + String.valueOf(sendResult));
		} catch (MQBrokerException e) {
			logger.error("MQBrokerException" + e.getMessage() + String.valueOf(sendResult));
		} catch (Exception e) {
			logger.error("Exception" + e.getMessage() + String.valueOf(sendResult));
		}
		/******************* 消息发送失败处理流程start **************************/
		if (sendResult == null || sendResult.getSendStatus() != SendStatus.SEND_OK) {
			logger.error("消息发送失败！！！！" + sendResult);
		}
		/******************* 消息发送失败处理流程end **************************/

	}

	/**
	 * @description:延时消息发送
	 * 1s,5s,10s,30s,1m......
	 * @param msg
	 * @date:2019年2月23日 下午1:59:58
	 * @author:wangyk3
	 */
	public void sendMsg(Message msg, int leval) {
		SendResult sendResult = null;
		try {
			msg.setDelayTimeLevel(leval);
			sendResult = producer.send(msg);
			logger.info(sendResult.toString());
		} catch (InterruptedException e) {
			logger.error("InterruptedException" + e.getMessage() + String.valueOf(sendResult));
		} catch (RemotingException e) {
			logger.error("RemotingException" + e.getMessage() + String.valueOf(sendResult));
		} catch (MQClientException e) {
			logger.error("MQClientException" + e.getMessage() + String.valueOf(sendResult));
		} catch (MQBrokerException e) {
			logger.error("MQBrokerException" + e.getMessage() + String.valueOf(sendResult));
		} catch (Exception e) {
			logger.error("Exception" + e.getMessage() + String.valueOf(sendResult));
		}
		/******************* 消息发送失败处理流程start **************************/
		if (sendResult == null || sendResult.getSendStatus() != SendStatus.SEND_OK) {
			logger.error("消息发送失败！！！！" + sendResult);
		}
		/******************* 消息发送失败处理流程end **************************/

	}

	public DefaultMQProducer getProducer() {
		return producer;
	}

	public void setProducerGroup(String producerGroup) {
		this.producerGroup = producerGroup;
	}

	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}

	public String getRetryAnotherBrokerWhenNotStoreOK() {
		return retryAnotherBrokerWhenNotStoreOK;
	}

	public void setRetryAnotherBrokerWhenNotStoreOK(String retryAnotherBrokerWhenNotStoreOK) {
		this.retryAnotherBrokerWhenNotStoreOK = retryAnotherBrokerWhenNotStoreOK;
	}

}
